package com.ruoyi.common.utils.rabbitmq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author xudengyu
 */
@Service
@Slf4j
public class RabbitMQConsumerService {

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;

    // 存储队列与对应监听容器的映射，用于管理和销毁
    private final ConcurrentMap<String, DirectMessageListenerContainer> containerMap = new ConcurrentHashMap<>();

    /**
     * 添加队列监听
     * @param queueName 队列名称
     * @param userId 用户ID（用于WebSocket消息路由）
     */
    public void addQueueListener(String queueName, String userId) {
        // 原子性判断队列是否已注册，避免重复监听
        if (containerMap.containsKey(queueName)) {
            log.warn("队列 {} 已被监听，无需重复注册", queueName);
            return;
        }

        // 创建消息监听器，处理消息并绑定用户ID
        ChannelAwareMessageListener listener = (message, channel) -> {
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            try {
                String payload = new String(message.getBody());
                String destination = "/topic/user." + userId;
                simpMessagingTemplate.convertAndSend(destination, payload);
                log.debug("成功消费消息 [队列: {}, DeliveryTag: {}]", queueName, deliveryTag);
                channel.basicAck(deliveryTag, false); // 手动确认消息
            } catch (Exception e) {
                log.error("处理消息失败 [队列: {}, DeliveryTag: {}]，消息将重新入队", queueName, deliveryTag, e);
                channel.basicNack(deliveryTag, false, true); // 处理失败，重新入队
            }
        };

        // 创建独立的监听容器
        DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
        container.setQueueNames(queueName);
        container.setMessageListener(listener);
        container.setConsumersPerQueue(1); // 每个队列1个消费者
        container.setAutoStartup(true); // 自动启动
        container.start();

        // 将容器存入映射表
        containerMap.put(queueName, container);
        log.info("成功注册队列监听: {}", queueName);
    }

    /**
     * 移除队列监听
     * @param queueName 队列名称
     */
    public void removeQueueListener(String queueName) {
        DirectMessageListenerContainer container = containerMap.remove(queueName);
        if (container != null) {
            container.stop(); // 停止监听
            log.info("已停止监听队列: {}", queueName);
        }
    }

    /**
     * 应用关闭时清理所有监听器
     */
    @PreDestroy
    public void destroy() {
        containerMap.forEach((queueName, container) -> {
            container.stop();
            log.info("应用关闭，停止监听队列: {}", queueName);
        });
        containerMap.clear();
    }
}